<?php

namespace houze;

use houze\kafka\Config\SslConfig;

use houze\kafka\Producer\Producer;
use houze\kafka\Producer\ProducerConfig;
use houze\kafka\Protocol\RecordBatch\RecordHeader;

use houze\kafka\Consumer\ConsumeMessage;
use houze\kafka\Consumer\Consumer;
use houze\kafka\Consumer\ConsumerConfig;

/*
* Kafka操作类 https://github.com/swoole/phpkafka
*/
class Kafka{
    // 生产者配置
    private static $producer_config = null;
    
    // 消费者配置
    private static $consumer_config = null;
    
    // 生产者实例
    private static $producer = null;
    
    // 消费者实例
    private static $consumer = null;
    
    // kafka配置
    private static $kafka_config = null;
    
    // 初始化
    public function __construct(){
        // Kafka
        if(empty(self::$kafka_config)) self::$kafka_config = \houze\Config::get('kafka.default');
        // 生产者
        if(empty(self::$producer_config)) self::$producer_config = new ProducerConfig();
        // 消费者
        if(empty(self::$consumer_config)) self::$consumer_config = new ConsumerConfig();
        
        // 生产者配置实例化
        self::$producer_config->setBootstrapServer(self::$kafka_config['bootstrap_servers']);
        self::$producer_config->setUpdateBrokers(self::$kafka_config['update_brokers']);
        self::$producer_config->setAcks(self::$kafka_config['acks']);
        // SASL支持
        self::$producer_config->setSasl(self::$kafka_config['sasl']);
        // SSL支持
        self::$producer_config->setSsl(self::$kafka_config['ssl']);
        self::$producer = new Producer(self::$producer_config);
        
        // 消费者配置实例化
        self::$consumer_config = new ConsumerConfig();
        self::$consumer_config->setBroker(self::$kafka_config['brokers']);
        // SASL支持
        self::$consumer_config->setSasl(self::$kafka_config['sasl']);
        // SSL支持
        self::$consumer_config->setSsl(self::$kafka_config['ssl']);
    }
    
    // 投递
    public function push($topic, $data, $headers=null){
        // 是否开启头文件
        if(empty($headers)){
            foreach ($data as $key => $value){
                self::$producer->send($topic, $value, $key);
            }
        }else{
            foreach ($data as $key => $value){
                self::$producer->send($topic, $value, $key, $headers);
            }
        }
    }
    
    // 消费
    public function pop($num=1, $topic, $group_id, $client_id){
        if($task) self::$consumer_config->setInterval(self::$kafka_config['interval']);
        self::$consumer_config->setTopic($topic); // 主题名称
        self::$consumer_config->setGroupId($group_id); // 消费分组ID
        self::$consumer_config->setClientId($client_id); // 消费者ID，不同的消费者进程请使用不同的设置
        self::$consumer_config->setGroupInstanceId($client_id); // 消费分组实例ID，不同的消费者进程请使用不同的设置
        
        // 判断消费次数 0为完全消费
        if($num === 0){
            // 实例化异步消费
            self::$consumer = new Consumer(self::$consumer_config, 'consume');
            // 异步消费
            self::$consumer->start();
        }else {
            // 实例化同步消费
            self::$consumer = new Consumer($consumer_config);
            for($i=0;$i>=$num;$i++){
                $message = self::$consumer->consume();
                if($message) {
                    var_dump($message->getKey() . ':' . $message->getValue());
                    self::$consumer->ack($message); // 手动提交
                }
                sleep(1);
            }
        }
    }
    
    // 异步任务回调
    private function consume(ConsumeMessage $message){
        var_dump($message->getKey() . ':' . $message->getValue());
        // $consumer->ack($message); // autoCommit设为false时，手动提交
    }
    
}